#ifndef __M_MSG_H__
#define __M_MSG_H__
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <iostream>
#include <string>
#include <unordered_map>
#include <memory>
#include <mutex>
#include <list>

// 定义消息类
namespace csjmq
{
#define DATAFILE_SUBFIX ".mqd"
#define TMPFILE_SUBFIX ".mqd.tmp"
    using MessagePtr = std::shared_ptr<csjmq::Message>;
    class MessageMapper
    {
    public:
        MessageMapper(std::string &basedir, const std::string &qname)
            : _qname(qname)
        {
            if (basedir.back() != '/')
                basedir.push_back('/');
            // 例如 test/queue1.mqd
            _datafile = basedir + qname + DATAFILE_SUBFIX;
            _tmpfile = basedir + qname + TMPFILE_SUBFIX;
            if (FileHelper(basedir).exists() == false)
                assert(FileHelper::createDirectory(basedir));
            assert(createMsgFile());
        }

        bool createMsgFile()
        {
            // 文件如果存在了，就不要再创建了
            if (FileHelper(_datafile).exists() == true)
                return true;
            bool ret = FileHelper::createFile(_datafile);
            if (ret == false)
            {
                ELOG("创建队列数据文件 %s 失败", _datafile.c_str());
                return false;
            }
            return true;
        }

        void removeMsgFile()
        {
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tmpfile);
        }

        bool insert(MessagePtr &msg)
        {
            return insert(_datafile, msg);
        }

        bool remove(MessagePtr &msg)
        {
            // 1.将msg的有效标志位修改为"0"
            msg->mutable_payload()->set_valid("0");

            // 2.将msg进行序列化
            std::string body = msg->payload().SerializeAsString();
            if (body.size() != msg->length())
            {
                DLOG("不能修改文件信息，因为新数据长度和原数据长度不一致");
                return false;
            }

            // 3.将序列化后的消息，写入数据文件的指定位置（覆盖旧数据）
            FileHelper helper(_datafile);
            bool ret = helper.write(body.c_str(), msg->offset(), body.size());
            if (ret == false)
            {
                DLOG("向队列数据文件写入数据失败");
                return false;
            }
            return true;
        }

        // 垃圾回收，重新加载有效数据
        std::list<MessagePtr> gc()
        {
            bool ret;
            std::list<MessagePtr> result;
            // 1.加载有效数据
            ret = load(result);
            if (ret == false)
            {
                DLOG("加载有效数据失败");
                return result;
            }

            // DLOG("加载有效数据的数量：%ld", result.size());
            // 2.将有效数据写入临时文件
            FileHelper::createFile(_tmpfile);
            for (auto &msg : result)
            {
                // DLOG("写入临时数据：%s", msg->payload().body().c_str());
                ret = insert(_tmpfile, msg);
                if (ret == false)
                {
                    DLOG("数据写入临时文件失败");
                    return result;
                }
            }

            // 3.删除源文件
            ret = FileHelper::removeFile(_datafile);
            if (ret == false)
            {
                DLOG("删除源文件失败");
                return result;
            }

            // 4.修改临时文件名为源文件名
            ret = FileHelper(_tmpfile).rename(_datafile);
            if (ret == false)
            {
                DLOG("重命名临时文件名为源文件名失败");
                return result;
            }

            // 5.返回新的有效数据
            return result;
        }

    private:
        bool load(std::list<MessagePtr> &result)
        {
            // 文件中消息的格式：四字节长度|数据|四字节长度|数据
            size_t offset = 0, msg_size;
            FileHelper data_file_helper(_datafile);
            size_t fsize = data_file_helper.size();
            bool ret;
            while (offset < fsize)
            {
                ret = data_file_helper.read((char *)&msg_size, offset, sizeof(size_t));
                if (ret == false)
                {
                    DLOG("读取消息长度失败");
                    return false;
                }
                offset += sizeof(size_t);
                std::string msg_body(msg_size, '\0');
                ret = data_file_helper.read(&msg_body[0], offset, msg_size);
                if (ret == false)
                {
                    DLOG("读取消息内容失败");
                    return false;
                }
                offset += msg_size;
                MessagePtr msgp = std::make_shared<Message>();
                msgp->mutable_payload()->ParseFromString(msg_body);
                // 将读取到的文件内容进行序列化，再判断是否有效
                if (msgp->payload().valid() == "0")
                {
                    DLOG("加载到无效消息：%s", msgp->payload().body().c_str());
                    continue;
                }
                result.push_back(msgp);
            }
            return true;
        }

        bool insert(const std::string &filename, MessagePtr &msg)
        {
            // 新增数据都是在文件的末尾的，刚开始文件末尾的位置是0
            // 1.进行消息的序列化，获取到格式化的消息
            std::string body = msg->payload().SerializeAsString();

            // 2.获取文件的长度
            FileHelper helper(filename);
            size_t fsize = helper.size();
            size_t msg_size = body.size();
            // 写入逻辑：1.先写4字节长度  2.后写入数据
            // 3.将数据写入到指定的位置
            bool ret = helper.write((char *)&msg_size, fsize, sizeof(size_t));
            if (ret == false)
            {
                ELOG("向队列数据文件中写入数据长度失败");
                return false;
            }

            ret = helper.write(body.c_str(), fsize + sizeof(size_t), body.size());
            if (ret == false)
            {
                ELOG("向队列数据文件中写入数据失败");
                return false;
            }

            // 4.更新msg中实际的存储位置
            msg->set_offset(fsize + sizeof(size_t));
            msg->set_length(body.size());
            return true;
        }

    private:
        // 依靠队列名称生成对应的队列文件和临时文件
        std::string _qname;
        std::string _datafile;
        std::string _tmpfile;
    };

    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;
        QueueMessage(std::string &basedir, const std::string &qname)
            : _mapper(basedir, qname), _qname(qname), _valid_count(0), _total_count(0)
        {
        }
        void recovery()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _msgs = _mapper.gc();
            for (auto &msg : _msgs)
            {
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            _valid_count = _total_count = _msgs.size();
        }

        bool insert(const BasicProperties *bp, const std::string body, bool queue_is_durable)
        {
            // 1.构造消息对象
            MessagePtr msg = std::make_shared<Message>();
            // 新增消息 = 属性 + 内容
            msg->mutable_payload()->set_body(body);
            if (bp != nullptr)
            {
                // 如果队列是持久化存储的，那么就默认消息的存储方式
                DeliveryMode mode = queue_is_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                DeliveryMode mode = queue_is_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());
                msg->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }

            std::unique_lock<std::mutex> lock(_mutex);
            // 2.判断消息是否持久化
            if (msg->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                msg->mutable_payload()->set_valid("1"); // 在持久化存储中表示数据有效
                // 3.进行持久化存储
                bool ret = _mapper.insert(msg);
                if (ret == false)
                {
                    DLOG("持久化存储 %s 消息失败", body.c_str());
                    return false;
                }
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                _total_count += 1;
                _valid_count += 1;
            }

            // 4.内存管理
            _msgs.push_back(msg);
            return true;
        }

        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_msgs.size() == 0)
            {
                return MessagePtr();
            }
            // 获取_msgs中链表头的数据
            MessagePtr msg = _msgs.front();
            _msgs.pop_front();
            // 将该数据添加到带确认消息map中
            _waitask_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            return msg;
        }

        // 移除一个消息后，内容确认是否需要进行垃圾回收
        bool remove(const std::string &msg_id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 1.待确认队列中查找消息
            auto it = _waitask_msgs.find(msg_id);
            if (it == _waitask_msgs.end())
            {
                DLOG("待确认消息队列中没有该消息：%s", msg_id.c_str());
                return true;
            }
            // 2.根据消息持久化模式，决定是否删除持久化信息
            if (it->second->payload().properties().delivery_mode() == DeliveryMode::DURABLE)
            {
                // 3.删除持久化
                _mapper.remove(it->second);
                _durable_msgs.erase(msg_id);
                _valid_count -= 1;
                gc(); // 删除一个消息后，判断是否需要进行垃圾回收
            }

            // 4.内存删除持久化;
            _waitask_msgs.erase(msg_id);
            return true;
        }

        size_t getable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }

        size_t total_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }

        size_t durable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }

        size_t waitack_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitask_msgs.size();
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeMsgFile();
            _msgs.clear();
            _waitask_msgs.clear();
            _durable_msgs.clear();
            _valid_count = 0;
            _total_count = 0;
        }

    private:
        // 垃圾回收
        void gc()
        {
            if (GCCheck() == false)
                return;
            std::list<MessagePtr> msgs = _mapper.gc();
            for (auto &msg : msgs)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if (it == _durable_msgs.end())
                {
                    DLOG("垃圾回收后，有一条持久化消息，在内存中没有进行管理!");
                    _msgs.push_back(msg);
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                    continue;
                }

                it->second->set_offset(msg->offset());
                it->second->set_length(msg->length());
            }
            _valid_count = _total_count = msgs.size();
        }

        bool GCCheck()
        {
            // 当消息的总数量大于2000且有效消息数量在总消息数量的50%以下
            if (_total_count > 2000 && _valid_count * 10 / _total_count < 5)
            {
                return true;
            }
            return false;
        }

    private:
        std::mutex _mutex;
        std::string _qname;
        size_t _valid_count;
        size_t _total_count;
        MessageMapper _mapper;
        std::list<MessagePtr> _msgs;                               // 待推送消息
        std::unordered_map<std::string, MessagePtr> _waitask_msgs; // 待确认消息map
        std::unordered_map<std::string, MessagePtr> _durable_msgs; // 持久化消息map
    };

    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>;
        MessageManager(const std::string basedir) : _basedir(basedir) {}

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto &msg : _queue_msgs)
            {
                msg.second->clear();
            }
        }
        void initQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it != _queue_msgs.end())
                    return;
                qmp = std::make_shared<QueueMessage>(_basedir, qname);
                _queue_msgs.insert(std::make_pair(qname, qmp));
            }
            qmp->recovery();
        }

        void destroyQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                    return;
                qmp = it->second;
                _queue_msgs.erase(it);
            }
            qmp->clear();
        }

        bool insert(const std::string &qname, const BasicProperties *bp, const std::string body,
                    bool queue_is_durable)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("向队列 %s 新增消息失败，没有找到对应的消息管理句柄", qname.c_str());
                    return false;
                }
                qmp = it->second;
            }
            return qmp->insert(bp, body, queue_is_durable);
        }

        MessagePtr front(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列 %s 对首消息失败，没有找到对应的消息管理句柄", qname.c_str());
                    return MessagePtr();
                }
                qmp = it->second;
            }
            return qmp->front();
        }

        void ack(const std::string &qname, const std::string &msg_id)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("确认队列 %s 消息失败，没有找到对应的消息管理句柄", qname.c_str());
                    return;
                }
                qmp = it->second;
            }
            qmp->remove(msg_id);
            return;
        }

        size_t getable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列 %s 待推送消息数量失败，没有找到对应的消息管理句柄", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->getable_count();
        }

        size_t total_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列 %s 总持久化消息数量失败，没有找到对应的消息管理句柄", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->total_count();
        }

        size_t durable_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列 %s 有效持久化数量失败，没有找到对应的消息管理句柄", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->durable_count();
        }

        size_t waitack_count(const std::string &qname)
        {
            QueueMessage::ptr qmp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end())
                {
                    DLOG("获取队列 %s 待确认消息数量失败，没有找到对应的消息管理句柄", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->waitack_count();
        }

    private:
        std::mutex _mutex;
        std::string _basedir;
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msgs;
    };
}
#endif
